package com.meiya.flink.service;

import com.meiya.flink.config.AppConfig;
import com.meiya.flink.interfaces.TagProcessor;
import com.meiya.flink.schema.TableDefinitions;
import com.meiya.flink.util.AsyncSqlUtil;
import com.meiya.flink.util.ClassLoaderUtil;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.kafka.clients.admin.AdminClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Modifier;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

/**
 * 标签处理服务
 */
public class TagProcessorService {
    private static final Logger logger = LoggerFactory.getLogger(TagProcessorService.class);

    private final StreamTableEnvironment tableEnv;
    private AdminClient kafkaAdminClient;
    private final Map<String, TagProcessor> tagProcessors = new HashMap<>();
    private boolean isRunning = false;

    public TagProcessorService(StreamTableEnvironment tableEnv) {
        this.tableEnv = tableEnv;
        initKafkaAdminClient();
        registerTagProcessors();
    }

    /**
     * 注册标签处理器
     */
    private void registerTagProcessors() {
        try {
            // 获取processor包路径
            String packageName = "com.meiya.flink.processor";

            // 使用反射获取所有TagProcessor实现类
            Set<Class<?>> classes = ClassLoaderUtil.getClasses(packageName);

            for (Class<?> clazz : classes) {
                // 判断是否是TagProcessor的实现类且不是接口或抽象类
                if (TagProcessor.class.isAssignableFrom(clazz)
                        && !Modifier.isInterface(clazz.getModifiers())
                        && !Modifier.isAbstract(clazz.getModifiers())) {

                    try {
                        // 实例化处理器
                        TagProcessor processor = (TagProcessor) clazz.getDeclaredConstructor().newInstance();

                        // 参数校验
                        if (processor.getTagName() == null || processor.getTagName().trim().isEmpty()) {
                            logger.warn("标签处理器校验失败: {}, 原因: tagName不能为空", clazz.getName());
                            continue;
                        }

                        // 检查是否已存在相同的标签名
                        if (tagProcessors.containsKey(processor.getTagName())) {
                            logger.warn("标签处理器校验失败: {}, 原因: 标签名'{}'已存在", clazz.getName(), processor.getTagName());
                            continue;
                        }

                        // 注册处理器
                        tagProcessors.put(processor.getTagName(), processor);
                        logger.debug("成功注册标签处理器: {}", clazz.getSimpleName());
                    } catch (Exception e) {
                        logger.error("注册标签处理器失败: {}, 原因: {}", clazz.getName(), e.getMessage());
                    }
                }
            }

            logger.info("已注册{}个标签处理器: {}", tagProcessors.size(), tagProcessors.keySet());
        } catch (Exception e) {
            logger.error("扫描注册标签处理器时发生错误: {}", e.getMessage());
        }
    }

    /**
     * 初始化Kafka AdminClient
     * 使用自定义配置，避免JMX MBean注册冲突
     */
    private void initKafkaAdminClient() {
        try {
            Properties adminProps = AppConfig.getKafkaAdminProperties();
            this.kafkaAdminClient = AdminClient.create(adminProps);
            logger.info("TagProcessorService: Kafka AdminClient初始化成功");
        } catch (Exception e) {
            logger.warn("TagProcessorService: Kafka AdminClient初始化失败: {}", e.getMessage());
        }
    }

    /**
     * 启动实时监听Kafka推送数据并处理标签
     */
    public void startRealTimeProcessing() {
        if (isRunning) {
            logger.warn("实时处理已经在运行中");
            return;
        }

        isRunning = true;
        logger.info("开始实时监听Kafka推送数据...");

        try {
            // 设置创建Kafka交易源表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建Kafka交易源表");
            logger.info("设置创建Kafka交易源表作业名称");
            
            // 创建Kafka交易源表
            String createKafkaSourceTable = TableDefinitions.getCreateKafkaSourceTableSQL();
            logger.info("创建Kafka交易源表SQL: {}", createKafkaSourceTable);
            try {
                tableEnv.executeSql(createKafkaSourceTable);
                logger.info("Kafka交易源表创建成功");
            } catch (Exception e) {
                logger.error("创建Kafka交易源表失败: {}", e.getMessage());
                return;
            }

            // 设置创建Kafka账户源表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建Kafka账户源表");
            logger.info("设置创建Kafka账户源表作业名称");
            
            // 创建Kafka账户源表
            String createKafkaAccountSourceTable = TableDefinitions.getCreateKafkaAccountSourceTableSQL();
            logger.info("创建Kafka账户源表SQL: {}", createKafkaAccountSourceTable);
            try {
                tableEnv.executeSql(createKafkaAccountSourceTable);
                logger.info("Kafka账户源表创建成功");
            } catch (Exception e) {
                logger.error("创建Kafka账户源表失败: {}", e.getMessage());
                return;
            }

            // 设置创建Doris交易明细表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建Doris交易明细表");
            logger.info("设置创建Doris交易明细表作业名称");
            
            // 创建Doris交易明细表
            String createDorisTransactionTable = TableDefinitions.getCreateDorisTransactionTableSQL();
            logger.info("创建Doris交易明细表SQL: {}", createDorisTransactionTable);
            tableEnv.executeSql(createDorisTransactionTable);
            logger.info("Doris交易明细表创建成功");

            // 设置创建Doris账户表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建Doris账户表");
            logger.info("设置创建Doris账户表作业名称");
            
            // 创建Doris账户表
            String createDorisAccountTable = TableDefinitions.getCreateDorisAccountTableSQL();
            logger.info("创建Doris账户表SQL: {}", createDorisAccountTable);
            tableEnv.executeSql(createDorisAccountTable);
            logger.info("Doris账户表创建成功");

            // 设置创建标签结果表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建Kafka标签结果表");
            logger.info("设置创建Kafka标签结果表作业名称");
            
            // 创建标签结果表
            String createTagsResultTable = TableDefinitions.getCreateTagsResultTableSQL();
            logger.info("创建Kafka标签结果表SQL: {}", createTagsResultTable);
            tableEnv.executeSql(createTagsResultTable);
            logger.info("Kafka标签结果表创建成功");

            // 设置创建Doris标签结果表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建Doris标签结果表");
            logger.info("设置创建Doris标签结果表作业名称");
            
            // 创建Doris标签结果表
            String createDorisTagsResultTable = TableDefinitions.getCreateDorisTagsResultTableSQL();
            logger.info("创建Doris标签结果表SQL: {}", createDorisTagsResultTable);
            tableEnv.executeSql(createDorisTagsResultTable);
            logger.info("Doris标签结果表创建成功");

            // 设置创建已处理交易表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建已处理交易表");
            logger.info("设置创建已处理交易表作业名称");
            
            // 创建已处理交易表
            String createProcessedTradesTable = TableDefinitions.getCreateProcessedTradesTableSQL();
            logger.info("创建已处理交易表SQL: {}", createProcessedTradesTable);
            tableEnv.executeSql(createProcessedTradesTable);
            logger.info("已处理交易表创建成功");

            // 设置创建已处理标签表的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建已处理标签表");
            logger.info("设置创建已处理标签表作业名称");
            
            // 创建已处理标签表
            String createProcessedTagsTable = TableDefinitions.getCreateProcessedTagsTableSQL();
            logger.info("创建已处理标签表SQL: {}", createProcessedTagsTable);
            tableEnv.executeSql(createProcessedTagsTable);
            logger.info("已处理标签表创建成功");

            // 创建实时标签计算SQL
            createRealTimeTagProcessingJob();

            logger.info("实时监听Kafka推送数据已启动");
        } catch (Exception e) {
            isRunning = false;
            logger.error("启动实时监听Kafka推送数据失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 创建实时标签计算作业
     * 基于会话窗口，当窗口关闭时触发标签计算
     */
    private void createRealTimeTagProcessingJob() {
        logger.info("创建实时标签计算作业...");

        try {
            // 检查必要的表是否存在
            boolean processedTradesTableExists = AsyncSqlUtil.checkTableExists("processed_trades", tableEnv::executeSql);
            boolean processedTagsTableExists = AsyncSqlUtil.checkTableExists("processed_tags", tableEnv::executeSql);

            // 创建基础视图，这些视图无论是否存在中间表都需要创建
            createBaseViews();

            // 只有当两个中间表都存在时才进行标签SQL处理
            if (processedTradesTableExists && processedTagsTableExists) {
                // 为每个标签创建实时计算SQL
                for (TagProcessor processor : tagProcessors.values()) {
                    String tagName = processor.getTagName();
                    long mid = processor.getMId();
                    logger.info("开始创建{}标签的实时计算SQL", tagName);

                    try {
                        // 生成标签计算SQL
                        String tagSQL = processor.generateTagSQL();
                        logger.info("{}标签的实时计算SQL: {}", tagName, tagSQL);

                        // 设置任务名称为标签名称
                        tableEnv.getConfig().set("pipeline.name", "标签计算-" + tagName + "-" + processor.getMId());
                        logger.info("设置标签计算作业名称: {}", tagName);

                        // 提交SQL执行
                        try {
                            tableEnv.executeSql(tagSQL);
                            logger.info("{}标签的实时计算SQL已提交", tagName);
                        } catch (Exception e) {
                            logger.error("{}标签的实时计算SQL执行失败: {}", tagName, e.getMessage(), e);
                            // 对于失败的标签，记录但不中断其他标签的处理
                            continue;
                        }

                        // 设置Doris任务名称
                        tableEnv.getConfig().set("pipeline.name", "Doris标签计算-" + tagName + "-" + processor.getMId());
                        logger.info("设置Doris标签计算作业名称: {}", tagName);

                        // 根据标签类型决定是否需要写入Doris
                        // 对于夜间高频标签，采用延迟写入Doris策略，避免直接查询超时
                        if (mid == 1002L) { // 夜间高频标签
                            // 使用简化版的SQL向Doris写入，避免复杂查询
                            String simplifiedDorisSQL = "INSERT INTO doris_tags_result\n" +
                                    "SELECT *\n" +
                                    "FROM tags_result\n" +
                                    "WHERE mid = " + mid + "\n" +
                                    "AND tags = '" + tagName + "'";
                            
                            try {
                                tableEnv.executeSql(simplifiedDorisSQL);
                                logger.info("{}标签的简化Doris写入SQL已提交", tagName);
                            } catch (Exception e) {
                                logger.error("{}标签的简化Doris写入SQL执行失败: {}", tagName, e.getMessage(), e);
                            }
                        } else {
                            // 其他标签直接应用原SQL替换目标表
                            String dorisTagSQL = tagSQL.replace("INSERT INTO tags_result", "INSERT INTO doris_tags_result");
                            try {
                                tableEnv.executeSql(dorisTagSQL);
                                logger.info("{}标签的Doris实时计算SQL已提交", tagName);
                            } catch (Exception e) {
                                logger.error("{}标签的Doris实时计算SQL执行失败: {}", tagName, e.getMessage(), e);
                            }
                        }
                    } catch (Exception e) {
                        logger.error("处理{}标签时发生错误: {}", tagName, e.getMessage(), e);
                    }
                }

                // 更新已处理记录表
                updateProcessedTables();

                logger.info("实时标签计算作业创建完成");
            } else {
                // 记录错误信息
                logger.error("无法创建标签计算SQL: processed_trades表存在: {}, processed_tags表存在: {}",
                        processedTradesTableExists, processedTagsTableExists);
                logger.error("标签计算需要两个中间表都存在才能进行");
            }
        } catch (Exception e) {
            logger.error("创建实时标签计算作业失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 更新已处理记录表
     */
    private void updateProcessedTables() {
        try {
            String[] updateSQLs = createUpdateProcessedTablesSQL();
            
            // 设置更新processed_trades表的作业名称
            tableEnv.getConfig().set("pipeline.name", "更新已处理交易表");
            logger.info("设置更新已处理交易表作业名称");

            // 更新processed_trades表
            logger.info("提交实时更新processed_trades表的SQL");
            try {
                tableEnv.executeSql(updateSQLs[0]);
                logger.info("实时更新processed_trades表的SQL已提交");
            } catch (Exception e) {
                logger.error("更新processed_trades表失败: {}", e.getMessage(), e);
            }

            // 设置更新processed_tags表的作业名称
            tableEnv.getConfig().set("pipeline.name", "更新已处理标签表");
            logger.info("设置更新已处理标签表作业名称");

            // 更新processed_tags表
            logger.info("提交实时更新processed_tags表的SQL");
            try {
                tableEnv.executeSql(updateSQLs[1]);
                logger.info("实时更新processed_tags表的SQL已提交");
            } catch (Exception e) {
                logger.error("更新processed_tags表失败: {}", e.getMessage(), e);
            }
        } catch (Exception e) {
            logger.error("更新已处理记录表失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 创建基础视图
     * 这些视图无论是否存在中间表都需要创建
     */
    private void createBaseViews() {
        try {
            // 创建或替换所有视图，使用try-catch分别处理每个视图创建，防止一个失败影响所有
            
            // 设置创建账户视图的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建账户信息视图");
            logger.info("设置创建账户信息视图作业名称");
            
            try {
                // 创建账户数据视图 - 直接从kafka源表分组，避免访问Doris
                String accountViewSQL = "CREATE VIEW IF NOT EXISTS account_info_view AS " +
                        "SELECT " +
                        "  clue_id, " +
                        "  jykh, " +
                        "  MAX(zhkhyh) AS zhkhyh, " +
                        "  task_record_id " +
                        "FROM account_source " +
                        "WHERE zhkhyh IS NOT NULL " +
                        "GROUP BY clue_id, jykh, task_record_id";

                logger.info("创建账户信息视图SQL: {}", accountViewSQL);
                tableEnv.executeSql(accountViewSQL);
                logger.info("账户信息视图创建成功");
            } catch (Exception e) {
                logger.error("创建账户信息视图失败: {}", e.getMessage(), e);
                // 创建简化后的账户视图，确保即使失败也有备选方案
                try {
                    String fallbackAccountViewSQL = "CREATE VIEW IF NOT EXISTS account_info_view AS " +
                            "SELECT " +
                            "  clue_id, " +
                            "  jykh, " +
                            "  '' AS zhkhyh, " +
                            "  task_record_id " +
                            "FROM transaction_source " +
                            "GROUP BY clue_id, jykh, task_record_id";
                    tableEnv.executeSql(fallbackAccountViewSQL);
                    logger.info("创建备用账户信息视图成功");
                } catch (Exception ex) {
                    logger.error("创建备用账户信息视图也失败: {}", ex.getMessage(), ex);
                }
            }

            // 获取会话窗口间隔时间
            int sessionWindowInterval = AppConfig.getSessionWindowIntervalMinutes();
            logger.info("当前会话窗口间隔时间: {} 分钟", sessionWindowInterval);

            // 设置创建交易与账户关联视图的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建交易与账户关联视图");
            logger.info("设置创建交易与账户关联视图作业名称");
            
            try {
                // 创建交易与账户关联视图，确保时间属性正确传递
                String joinedViewSQL = "CREATE VIEW IF NOT EXISTS transaction_account_view AS " +
                        "SELECT " +
                        "  t.clue_id, " +
                        "  t.jykh, " +
                        "  t.jyzh, " +
                        "  t.jyfhm, " +
                        "  t.jyfzjhm, " +
                        "  t.jysj, " +
                        "  t.jyje, " +
                        "  t.jyye, " +
                        "  t.sfbz, " +
                        "  t.jydszkh, " +
                        "  t.xjbz, " +
                        "  t.dshm, " +
                        "  t.dssfzh, " +
                        "  t.dskhyh, " +
                        "  t.zysm, " +
                        "  t.jybz, " +
                        "  t.jywdmc, " +
                        "  t.jyfsd, " +
                        "  t.jysfcg, " +
                        "  t.cph, " +
                        "  t.ipdz, " +
                        "  t.macdz, " +
                        "  t.dsjyye, " +
                        "  t.jylsh, " +
                        "  t.rzh, " +
                        "  t.pzzl, " +
                        "  t.pzh, " +
                        "  t.jygyh, " +
                        "  t.bz, " +
                        "  t.shmc, " +
                        "  t.jylx, " +
                        "  t.cxfkjgyy, " +
                        "  t.task_record_id, " +
                        "  t.a_yhlx, " +
                        "  t.action_type, " +
                        "  t.from_jykh, " +
                        "  t.to_jydszkh, " +
                        "  t.crrq, " +
                        "  t.ukey, " +
                        "  t.processing_time, " +
                        "  t.event_time, " +
                        "  CASE WHEN a.zhkhyh IS NOT NULL THEN a.zhkhyh " +
                        "       ELSE SUBSTR(t.a_yhlx, INSTR(t.a_yhlx, '-') + 1, 50) " +
                        "  END AS bank_name " +
                        "FROM transaction_source t " +
                        "LEFT JOIN account_info_view a ON t.clue_id = a.clue_id AND t.jykh = a.jykh " +
                        "WHERE (t.task_record_id = a.task_record_id OR (t.task_record_id IS NULL AND a.task_record_id IS NULL))";

                logger.info("创建交易与账户关联视图SQL: {}", joinedViewSQL);
                tableEnv.executeSql(joinedViewSQL);
                logger.info("交易与账户关联视图创建成功");
            } catch (Exception e) {
               logger.error("创建交易与账户关联视图失败: {}", e.getMessage(), e);
            }

            // 设置创建时间属性视图的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建时间属性视图");
            logger.info("设置创建时间属性视图作业名称");
            
            try {
                // 创建一个新的视图，显式声明时间属性
                String timeAttributeViewSQL = "CREATE VIEW IF NOT EXISTS transaction_time_attribute_view AS " +
                        "SELECT " +
                        "  *, " +
                        "  PROCTIME() AS proc_time " +
                        "FROM transaction_account_view";

                logger.info("创建时间属性视图SQL: {}", timeAttributeViewSQL);
                tableEnv.executeSql(timeAttributeViewSQL);
                logger.info("时间属性视图创建成功");
            } catch (Exception e) {
                logger.error("创建时间属性视图失败: {}", e.getMessage(), e);
            }

            // 设置创建会话窗口视图的作业名称
            tableEnv.getConfig().set("pipeline.name", "创建会话窗口视图");
            logger.info("设置创建会话窗口视图作业名称");
            
            try {
                // 创建会话窗口视图，使用处理时间作为时间属性
                String sessionWindowViewSQL = "CREATE VIEW IF NOT EXISTS transaction_session_window_view AS " +
                        "SELECT * FROM TABLE(" +
                        "  SESSION(TABLE transaction_time_attribute_view, " +
                        "          DESCRIPTOR(proc_time), " +
                        "          INTERVAL '" + sessionWindowInterval + "' MINUTE)" +
                        ") WHERE action_type = 0";

                logger.info("创建会话窗口视图SQL: {}", sessionWindowViewSQL);
                tableEnv.executeSql(sessionWindowViewSQL);
                logger.info("会话窗口视图创建成功");
            } catch (Exception e) {
                logger.error("创建会话窗口视图失败: {}", e.getMessage(), e);
            }
        } catch (Exception e) {
            logger.error("创建基础视图失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 创建实时更新processed_trades和processed_tags表的SQL
     *
     * @return 包含两个SQL语句的数组，第一个是更新processed_trades的SQL，第二个是更新processed_tags的SQL
     */
    private String[] createUpdateProcessedTablesSQL() {
        // 创建更新processed_trades表的SQL
        // 每个标签单独插入一条记录，使用标签ID作为mid
        StringBuilder tradesInsertBuilder = new StringBuilder();
        tradesInsertBuilder.append("INSERT INTO processed_trades\n");
        
        boolean firstTag = true;
        for (TagProcessor processor : tagProcessors.values()) {
            if (!firstTag) {
                tradesInsertBuilder.append("UNION ALL\n");
            }
            tradesInsertBuilder.append("SELECT\n")
                    .append("  t.clue_id,\n")
                    .append("  t.jykh,\n")
                    .append("  '").append(processor.getMId()).append("' AS mid,\n")
                    .append("  CURRENT_TIMESTAMP AS last_process_time\n")
                    .append("FROM (\n")
                    .append("  SELECT DISTINCT\n")
                    .append("    clue_id,\n")
                    .append("    jykh\n")
                    .append("  FROM transaction_account_view\n")
                    .append("  WHERE action_type = 0\n")
                    .append(") t\n")
                    .append("WHERE NOT EXISTS (\n")
                    .append("  SELECT 1 FROM processed_trades pt\n")
                    .append("  WHERE pt.clue_id = t.clue_id AND pt.jykh = t.jykh AND pt.mid = '").append(processor.getMId()).append("'\n")
                    .append(")\n");
            
            firstTag = false;
        }
        
        String tradesSQL = tradesInsertBuilder.toString();

        // 创建更新processed_tags表的SQL
        // 适配新的主键结构 (clue_id, jykh, mid)
        String tagsSQL = "INSERT INTO processed_tags\n" +
                "SELECT\n" +
                "  tr.clue_id,\n" +
                "  tr.jykh,\n" +
                "  CAST(tr.mid AS STRING) AS mid,\n" +
                "  CURRENT_TIMESTAMP AS last_process_time\n" +
                "FROM tags_result tr\n" +
                "WHERE NOT EXISTS (\n" +
                "  SELECT 1 FROM processed_tags pt\n" +
                "  WHERE pt.clue_id = tr.clue_id AND pt.jykh = tr.jykh AND pt.mid = CAST(tr.mid AS STRING)\n" +
                ")";

        return new String[]{tradesSQL, tagsSQL};
    }

    /**
     * 停止实时处理
     */
    public void stopRealTimeProcessing() {
        logger.info("停止实时处理...");
        isRunning = false;
    }

    /**
     * 查询标签结果
     */
    public void queryTagResults() {
        logger.info("开始查询标签结果...");

        try {
            // 设置查询标签结果的作业名称
            tableEnv.getConfig().set("pipeline.name", "查询标签结果");
            logger.info("设置查询标签结果作业名称");
            
            // 检查表是否存在
            boolean tableExists = AsyncSqlUtil.checkTableExists("tags_result", tableEnv::executeSql);
            if (!tableExists) {
                logger.error("标签结果表不存在，无法查询数据");
                return;
            }

            // 查询标签结果
            String sql = "SELECT * FROM tags_result LIMIT 10";

            logger.info("执行标签结果查询SQL: {}", sql);

            logger.info("标签结果查询成功: {}", sql);
            AsyncSqlUtil.executeAsyncWithThread(sql, tableEnv::executeSql, "标签结果查询");
            logger.info("标签结果查询END: {}", sql);

        } catch (Exception e) {
            logger.error("查询标签结果失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 清理资源
     */
    public void cleanup() {
        logger.info("开始清理资源...");

        try {
            // 停止实时处理
            stopRealTimeProcessing();

            // 中断所有活动线程
            AsyncSqlUtil.interruptAllThreads();

            // 关闭Kafka AdminClient
            if (kafkaAdminClient != null) {
                try {
                    logger.info("正在关闭Kafka AdminClient...");
                    kafkaAdminClient.close();
                    logger.info("Kafka AdminClient已关闭");
                } catch (Exception e) {
                    logger.warn("关闭Kafka AdminClient时发生错误: {}", e.getMessage());
                }
            }

            logger.info("资源清理完成");
        } catch (Exception e) {
            logger.error("资源清理失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 查询Kafka源表数据
     */
    public void queryKafkaSourceData() {
        logger.info("开始查询Kafka源表数据...");
        try {
            // 设置查询Kafka源表数据的作业名称
            tableEnv.getConfig().set("pipeline.name", "查询Kafka源表数据");
            logger.info("设置查询Kafka源表数据作业名称");
            
            // 检查表是否存在
            boolean tableExists = AsyncSqlUtil.checkTableExists("transaction_source", tableEnv::executeSql);
            if (!tableExists) {
                logger.error("Kafka源表不存在，无法查询数据");
                return;
            }

            // 简单的查询，限制结果数量
            String sql = "SELECT * FROM transaction_source LIMIT 10";
            logger.info("执行查询Kafka源表SQL: {}", sql);

            // 直接执行SQL并返回结果
            tableEnv.executeSql(sql).print();

            // 设置查询源表数量的作业名称
            tableEnv.getConfig().set("pipeline.name", "查询Kafka源表数量");
            logger.info("设置查询Kafka源表数量作业名称");
            
            // 查询源表数量
            String countSql = "SELECT COUNT(*) FROM transaction_source";
            logger.info("执行查询Kafka源表数量SQL: {}", countSql);
            tableEnv.executeSql(countSql).print();

            logger.info("Kafka源表数据查询完成");
        } catch (Exception e) {
            logger.error("查询Kafka源表数据失败: {}", e.getMessage(), e);
        }
    }

}